package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job08_各品牌支付额最大的前10个商品 {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        // 开启窗口的提前触发器（trigger）
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1000 ms");


        // 创建业务库中的oms_order表的cdc连接器表
        tenv.executeSql(
                " create table oms_order_mysql_cdc(                          \n "+
                        "   id           bigint,                             \n "+ // 订单id
                        "   status         int,                              \n "+ // 订单状态
                        " 	create_time  timestamp(3),                       \n "+  // 订单创建时间
                        " 	payment_time  timestamp(3),                      \n "+  // 订单支付时间
                        " 	modify_time  timestamp(3),                      \n "+
                        " 	rt as modify_time,                              \n "+
                        " 	watermark for rt as rt - interval '0' second,   \n "+
                        "   primary key (id) not enforced                   \n "+
                        " )WITH (                                           \n "+
                        "     'connector' = 'mysql-cdc',                    \n "+
                        "     'hostname' = 'doitedu',                       \n "+
                        "     'port' = '3306',                              \n "+
                        "     'username' = 'root',                          \n "+
                        "     'password' = 'root',                          \n "+
                        "     'database-name' = 'realtimedw',               \n "+
                        "     'table-name' = 'oms_order'                    \n "+
                        " )                                                 \n "
        );

        // 创建业务库中的 oms_order_item 的cdc连接器表
        tenv.executeSql(
                " create table oms_order_item_mysql_cdc(                    \n "+
                        "   id           bigint,                            \n "+ // 自增主键
                        "   order_id      bigint,                           \n "+ // 订单id
                        "   product_id    bigint,                           \n "+ // 商品id
                        "   product_brand  string,                          \n "+ // 商品品牌
                        " 	real_amount  decimal(10,2),                     \n "+  // 订单创建时间
                        "   primary key (id) not enforced                   \n "+
                        " )WITH (                                           \n "+
                        "     'connector' = 'mysql-cdc',                    \n "+
                        "     'hostname' = 'doitedu',                       \n "+
                        "     'port' = '3306',                              \n "+
                        "     'username' = 'root',                          \n "+
                        "     'password' = 'root',                          \n "+
                        "     'database-name' = 'realtimedw',               \n "+
                        "     'table-name' = 'oms_order_item'               \n "+
                        " )                                                 \n "
        );


        tenv.executeSql(
                        " CREATE TEMPORARY VIEW joined AS SELECT   "+
                        "     od.id as order_id,               "+
                        "     od.status,                       "+
                        "     od.create_time,                  "+
                        "     od.payment_time,                 "+
                        "     od.modify_time,                  "+
                        "     od.rt,                           "+
                        "     oi.product_id,                   "+
                        "     oi.product_brand,                "+
                        "     oi.real_amount                   "+
                        " from oms_order_mysql_cdc od          "+
                        " join oms_order_item_mysql_cdc oi     "+
                        " ON od.id = oi.order_id               "

        );

        tenv.executeSql("WITH tmp AS (\n" +
                "    SELECT\n" +
                "       *\n" +
                "    FROM joined \n" +
                "    WHERE date_format(payment_time,'yyyy-MM-dd') = CURRENT_DATE\n" +
                ")\n" +
                "\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    product_brand,\n" +
                "    product_id,\n" +
                "    real_amount,\n" +
                "\trn\n" +
                "FROM (\n" +
                "    SELECT\n" +
                "        window_start,\n" +
                "        window_end,\n" +
                "        product_brand,\n" +
                "        product_id,\n" +
                "        real_amount,\n" +
                "        row_number() over(partition by window_start,window_end,product_brand order by real_amount desc ) as rn\n" +
                "    \n" +
                "    FROM (\n" +
                "        SELECT\n" +
                "            TUMBLE_START(rt,interval '24' hour) as window_start,\n" +
                "            TUMBLE_END(rt,interval '24' hour) as window_end,\n" +
                "        \tproduct_brand,\n" +
                "        \tproduct_id,\n" +
                "            sum(real_amount) as real_amount\t\n" +
                "        FROM tmp \n" +
                "        GROUP BY \n" +
                "            TUMBLE(rt,interval '24' hour),\n" +
                "        \tproduct_brand,\n" +
                "        \tproduct_id\n" +
                "    ) o1 \n" +
                ") o2\n" +
                "WHERE rn<=10").print();






    }


}
